package com.qisi.functions;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.qisi.pojo.AppType;
import com.qisi.pojo.Rule;
import com.qisi.utils.Config;
import com.qisi.utils.MysqlCDCDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.io.IOException;
import java.sql.SQLException;
import java.util.Properties;

import static com.qisi.utils.Config.MYSQL_DIM_TABLE;
import static com.qisi.utils.Config.MYSQL_RULE_TABLE;

@Slf4j
public class SourceFunctions {

    public static FlinkKafkaConsumer<String> getKafkaSource() {
        Properties prop = new Properties();
        return new FlinkKafkaConsumer<>(Config.KAFKA_CONSUMER_TOPIC_LIST, new SimpleStringSchema(), prop);
    }


    /**
     * earliest、latest
     * @return Kafka 数据源，根据 APP_TYPE 确定读取的 topics
     */
    public static FlinkKafkaConsumer<String> getKafkaEventSource() throws IOException {

        Properties prop = new Properties();

        prop.load(SourceFunctions.class.getClassLoader().getResourceAsStream(Config.KAFKA_CONSUMER_CONFIG_FILE));

        log.info("consumer kafka topic is: {}, from: {}", Config.KAFKA_CONSUMER_TOPIC_LIST.toArray(), prop.getProperty("auto.offset.reset"));

        return new FlinkKafkaConsumer<>(Config.KAFKA_CONSUMER_TOPIC_LIST, new SimpleStringSchema(), prop);
    }

    /**
     * @return topic 自定义 Mysql 数据源，全量同步规则表
     */
    public static RichSourceFunction<Rule> getCustomMysqlSource(Configuration config) throws SQLException {

        return new CustomMysqlSourceFunction(config);

    }

    /**
     * Mysql 规则源数据源
     */
    public static SourceFunction<String> getMysqlRuleSource() {
        return getBinlogSourceFunction(MYSQL_RULE_TABLE);
    }

    /**
     * Mysql 维表源数据源
     */
    public static SourceFunction<String> getMysqlDimSource() {
        return getBinlogSourceFunction(MYSQL_DIM_TABLE);
    }

    /**
     * private 创建 MySQL binlog 数据源
     */
    private static SourceFunction<String> getBinlogSourceFunction(String table) {

        Properties properties = new Properties();
        properties.put("offset.flush.interval.ms", "10000");
        properties.put("offset.flush.timeout.ms", "40000");
        properties.put("autocommit", "false");

        return MySQLSource.<String>builder()
                .hostname(Config.MYSQL_HOST)
                .port(Config.MYSQL_PORT)
                .databaseList(Config.MYSQL_DB)
                .tableList(table)
                //.startupOptions(StartupOptions.latest())
                .username(Config.MYSQL_USER)
                .password(Config.MYSQL_PSWD)
                .deserializer(new MysqlCDCDeserializationSchema())
                .debeziumProperties(properties)
                .build();
    }
}
